﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Messaging;
using System.Runtime.Remoting.Messaging;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Transactions;

namespace Reflexive.IO
{
    public partial class MSMQReader<T> : IReflexive<T>, IDisposable 
    where T : class
    {

        MessageQueue _mq;
        SynchronizedCollection<Subscription<T>> _observers = new SynchronizedCollection<Subscription<T>>();

        public StatusCode Status
        {
            get;
            private set;
        }

        public MSMQReader(string queueName)
        {
            _mq = new MessageQueue(queueName, QueueAccessMode.ReceiveAndAdmin);
            {
                _mq.Formatter = new System.Messaging.XmlMessageFormatter(new Type[] { typeof(T) });
                _mq.ReceiveCompleted += new ReceiveCompletedEventHandler(ReceiveCompleted);
                _mq.PeekCompleted += new PeekCompletedEventHandler(PeekCompleted);
            }
        }

        public void Stop()
        {
            Status = StatusCode.Suspended;
        }

        public void Start()
        {
            if (Status == StatusCode.Completed)
            {
                throw new NotSupportedException("Reader is in the Completed state.");
            }
            Receive();
            Status = StatusCode.Started;
            
        }

        #region Queue Management

        private void Receive()
        {
            IAsyncResult results = _mq.BeginPeek();
        }

        private void ReceiveCompleted(object sender, ReceiveCompletedEventArgs args)
        {
            try
            {
                if (Status == StatusCode.Started)
                {
                    T message = ((T)args.Message.Body);
                    try
                    {
                        _observers.OnNext<T>(message);
                    }
                    catch (Exception e)
                    {
                        System.Diagnostics.Trace.TraceError(e.ToString());
                    }
                }

            }
            catch(Exception e)
            {
                System.Diagnostics.Trace.TraceError(e.ToString());
                try
                {
                    _observers.OnError<T>(e);
                }
                catch (Exception ex)
                {
                    System.Diagnostics.Trace.TraceError(ex.ToString());
                }
            }
            finally
            {
                if (Status == StatusCode.Started)
                {
                    Receive(); //call for next item in msmq
                }
            }
        }

        private void PeekCompleted(object sender, PeekCompletedEventArgs args)
        {
            try
            {
                if (Status == StatusCode.Started)
                {
                    using (var scope = new TransactionScope(TransactionScopeOption.Required))
                    {
                        try
                        {
                            T message = ((T)_mq.Receive(MessageQueueTransactionType.Automatic).Body);
                            
                            try
                            { 
                                 _observers.OnNext<T>(message);
                                 scope.Complete();
                            }
                            catch (Exception e)
                            {
                                System.Diagnostics.Trace.TraceError(e.ToString());
                            }
                        }
                        catch (Exception e)
                        {
                            System.Diagnostics.Trace.TraceError(e.ToString());
                        }
                    }
                   
                }

            }
            catch (Exception e)
            {
                System.Diagnostics.Trace.TraceError(e.ToString());
                try
                {
                    _observers.OnError<T>(e);
                }
                catch (Exception ex)
                {
                    System.Diagnostics.Trace.TraceError(ex.ToString());
                }
            }
            finally
            {
                if (Status == StatusCode.Started)
                {
                    Receive(); 
                }
            }
        }

        #endregion
       
        #region Subscription Management
    
        public IDisposable Subscribe(IObserver<T> observer)
        {
            var subscription = new Subscription<T>(observer);
            subscription.UnSubscribe += UnSubscribe;
            _observers.Add(subscription);
            return subscription;
        }

        private void UnSubscribe(object sender, IObserver<T> e)
        {
            _observers.Remove(((Subscription<T>)sender));
        }

        #endregion

        public void Dispose()
        {
            if (Status != StatusCode.Completed)
            {
                try
                {
                    if (_mq != null)
                    {
                        _mq.Close();
                        _mq.Dispose();
                    }
                    _observers.OnCompleted<T>();
                }
                catch (Exception e)
                {
                    System.Diagnostics.Trace.TraceError(e.ToString());
                }
                finally
                {
                    Status = StatusCode.Completed;
                }
            }
        }
    }
}
